package com.example.websocket.redis.config;

import com.example.websocket.redis.controller.RedisTestController;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;

import java.util.Map;

//@Component
public class RunRedisTestStreamReceiver extends AbstractRedisStreamReceiver<RedisGroupEvent>{
 
    private static final Logger LOGGER=LogManager.getLogger(RunRedisTestStreamReceiver.class);
    
    @Autowired
    private RedisTestController redisTestController;
    
    public RunRedisTestStreamReceiver() {
       super("test", "redis-test", "reactive-consumer-1");
    }
 
    protected Mono<Long> onMessage(ObjectRecord<String, RedisGroupEvent> message) {
        LOGGER.info("StreamReceiver received Stream[{}], event[{}]", message.getStream(), message.getValue());
        try {
            Map<String, Object> map = message.getValue().getData();
            String key = (String) map.get("key");
            String value = (String) map.get("value");
            redisTestController.runMethod(key, value);
            return redisTemplate.opsForStream().acknowledge(groupName, message);
        } catch (Exception e) {
            LOGGER.error("===========>运行异常！处理失败任务: {}", e, message.getValue());
            return redisTemplate.opsForStream().acknowledge(groupName, message);
        }
    }
    
}